package vip.zhenzicheng.rocketmq.rocketmq;

import org.apache.rocketmq.client.consumer.LitePullConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;

/**
 * @author zhenzicheng
 * @date 2022-10-09 11:19
 */
@Component
public class RocketMQConsumer {

  public static final Long MILLS_UNIT = 1000L;

  @Autowired
  LitePullConsumer litePullConsumer;

  /**
   * 主动轮询拉取消息
   * @param seconds 秒
   */
  public void poll(int seconds) {
    List<MessageExt> messageExts = litePullConsumer.poll(seconds * MILLS_UNIT);
    if (CollectionUtils.isEmpty(messageExts)) {
      for (MessageExt msg : messageExts) {
        byte[] body = msg.getBody();
        String msgStr = new String(body);
        // msg.getQ
      }
    }
    // litePullConsumer.committed()
  }

}
